package com.enhinck.dbutilweb.common.database;

import com.enhinck.dbutilweb.common.util.Database;
import com.enhinck.dbutilweb.common.util.JDBCUtil;
import com.enhinck.dbutilweb.model.domain.DatasourceConfDO;
import com.enhinck.dbutilweb.service.DatasourceConfService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.*;

/**
 * 描述
 *
 * @author huenbin
 * @date 3/3/21 7:37 PM
 */
@Component
@Slf4j
public class DbPoolUtil {
    public static Map<Long, Queue<ConnectionWrapper>> connectionWrapperMap = new ConcurrentHashMap();
    ScheduledExecutorService scheduledExecutorService;
    @Autowired
    MysqlDBUtil mysqlDBUtil;
    @Resource
    DatasourceConfService datasourceConfService;
    public static final long HOLD_TIME = 60_000;
    public static final long IDLE_TIME = 3600_000;
    public static final String HOLD_QUERY = "select 1 from dual";

    @PostConstruct
    public void init() {
        // 守护线程池
        scheduledExecutorService = new ScheduledThreadPoolExecutor(1,
                new BasicThreadFactory.Builder().namingPattern("dbConnection-schedule-pool-%d").daemon(true).build());
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            connectionWrapperMap.forEach((key, value) -> {
                if (value.size() > 0) {
                    final long current = System.currentTimeMillis();
                    // 保持会话并标记无效连接
                    value.parallelStream().forEach(connection -> {
                        if (connection.isAvailable()) {
                            holdConnection(current, connection);
                        }
                    });
                    // 清理无效连接
                    for (int i = 0; i < value.size(); i++) {
                        ConnectionWrapper connectionWrapper = value.poll();
                        if (connectionWrapper.isAvailable()) {
                            value.add(connectionWrapper);
                        }
                    }
                    log.info("id为{}的数据库连接池有:{}个连接数，当前执行会话保持功能...", key, value.size());
                }
            });
        }, 10, 30, TimeUnit.SECONDS);
    }

    /**
     * 获取数据源连接
     *
     * @param id
     * @return
     * @throws Exception
     */
    public ConnectionWrapper getConnection(Long id) throws Exception {
        Queue<ConnectionWrapper> queue = getQueue(id);

        if (queue.size() == 0) {
            return newConnection(id);
        } else {
            while (queue.size() > 0) {
                ConnectionWrapper connectionWrapper = queue.poll();
                if (connectionWrapper.isAvailable()) {
                    return connectionWrapper;
                }
            }
            return newConnection(id);
        }
    }


    /**
     * 回收连接
     *
     * @param connectionWrapper
     */
    public void release(ConnectionWrapper connectionWrapper) {
        if (connectionWrapper != null) {
            Queue<ConnectionWrapper> queue = getQueue(connectionWrapper.getId());
            queue.add(connectionWrapper);
        }
    }

    private Queue<ConnectionWrapper> getQueue(Long id) {
        Queue<ConnectionWrapper> queue = connectionWrapperMap.get(id);
        if (queue == null) {
            synchronized (this) {
                queue = connectionWrapperMap.get(id);
                if (queue == null) {
                    queue = new ConcurrentLinkedQueue<>();
                    connectionWrapperMap.put(id, queue);
                }
            }
        }
        return queue;
    }


    /**
     * 释放所有某数据源的链接
     *
     * @param id
     */
    public void closeAll(Long id) {
        Queue<ConnectionWrapper> queue = connectionWrapperMap.get(id);
        if (CollectionUtils.isNotEmpty(queue)) {
            while (queue.size() > 0) {
                ConnectionWrapper connectionWrapper = queue.poll();
                // 释放原来连接
                JDBCUtil.releaseConnection(connectionWrapper.getConnection());
            }
        }
    }

    /**
     * 创建新连接
     *
     * @param id
     * @return
     * @throws Exception
     */
    private ConnectionWrapper newConnection(Long id) throws Exception {
        ConnectionWrapper connectionWrapper = new ConnectionWrapper();
        connectionWrapper.setId(id);
        DatasourceConfDO datasourceConfDO = datasourceConfService.getById(id);
        Database database = new Database(datasourceConfDO.getUrl(), datasourceConfDO.getUsername(), datasourceConfDO.getPassword());
        Connection connection = database.getConnectionThrowEx();
        connectionWrapper.setLastQueryTime(System.currentTimeMillis());
        connectionWrapper.setConnection(connection);
        connectionWrapper.setDatabase(database);
        connectionWrapper.setAvailable(true);
        return connectionWrapper;
    }

    /**
     * 连接保持
     *
     * @param current
     * @param connection
     */
    private void holdConnection(long current, ConnectionWrapper connection) {
        if (current - connection.getLastHoldTime() >= HOLD_TIME) {
            try {
                if (current - connection.getLastHoldTime() >= IDLE_TIME) {
                    // 超过空闲时间 释放链接
                    connection.setAvailable(false);
                    JDBCUtil.releaseConnection(connection.getConnection());
                } else {
                    mysqlDBUtil.execQueryCommonThrowEx(connection.getConnection(), HOLD_QUERY);
                    connection.setLastHoldTime(current);
                    connection.setAvailable(true);
                }
            } catch (SQLException e) {
                log.info("数据库连接保持会话失败:{}", e);
                log.info("尝试重新连接...");
                // 释放原来连接
                JDBCUtil.releaseConnection(connection.getConnection());
                try {
                    Connection con = connection.getDatabase().getConnectionThrowEx();
                    connection.setConnection(con);
                    connection.setAvailable(true);
                } catch (Exception e1) {
                    log.info("尝试重新连接失败：{}", e1);
                    connection.setAvailable(false);
                    log.info("标记该连接失效");
                }
            }
        }
    }

}
